package com.example.java.state;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

// 计算平均值
public class ValueStateMain {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        DataStreamSource<Tuple2<Long, Long>> dataDs = env.fromElements(Tuple2.of(1L, 3L),
                Tuple2.of(1L, 5L), Tuple2.of(1L, 7L),
                Tuple2.of(1L, 4L), Tuple2.of(1L, 2L));
        KeyedStream<Tuple2<Long, Long>, Long> keyedStream = dataDs.keyBy(value -> value.f0);

        SingleOutputStreamOperator<Tuple2<Long, Long>> flatMappedStream = keyedStream
                .flatMap(new RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() {
                    private ValueState<Tuple2<Long, Long>> sumState;
                    private Long counter = 0L;

                    // 1.在open方法中初始化state
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 1.1 声明一个状态描述符，用于指定状态的属性（名称，类型信息，状态初始值）

                        ValueStateDescriptor<Tuple2<Long, Long>> descriptor = new ValueStateDescriptor<>(
                                "average",
                                TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {
                                }),
                                Tuple2.of(0L, 0L)
                        );
                        // 第二种声明状态描述符的方法(没初始值容易报null)
//                        ValueStateDescriptor<Tuple2<Long, Long>> descriptor2 = new ValueStateDescriptor<>(
//                                "average",
//                                TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {
//                                }));
                        // 1.2 根据state描述符创建一个valueState
                        sumState = getRuntimeContext().getState(descriptor);

                    }

                    // 2.在map方法中更新状态值
                    @Override
                    public void flatMap(Tuple2<Long, Long> value,
                                        Collector<Tuple2<Long, Long>> out) throws Exception {
                        //获取当前状态值
                        Tuple2<Long, Long> currentSum = sumState.value();
                        //更新
                        if (currentSum == null) {
                            System.out.println("null");
                        }
                        currentSum.f0 = ++counter;
                        currentSum.f1 += value.f1;
                        System.out.println("...currentSum:" + currentSum);

                        //更新状态值
                        sumState.update(currentSum);

//                        //如果count>=2 清空状态值，重新计算
//                        if (currentSum.f0 == 2) {
//                            out.collect(new Tuple2<>(counter, currentSum.f1 / currentSum.f0));
//                            sumState.clear();
//                        }
                        out.collect(new Tuple2<>(counter, currentSum.f1 / currentSum.f0));
                    }
                });

        flatMappedStream.print();
        env.execute();
    }
}
